package com.boarsoft.boar.batch.biz;

import java.sql.Timestamp;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.boarsoft.boar.batch.entity.BatchInst;
import com.boarsoft.boar.batch.entity.BatchLock;
import com.boarsoft.common.Util;
import com.boarsoft.common.biz.SimpleBizImpl;

@Component
public class BatchInstBizImpl extends SimpleBizImpl implements BatchInstBiz {

	@Autowired
	private BatchLockBiz batchLockBiz;

	@SuppressWarnings("unchecked")
	@Override
	public List<Object[]> list(String catalog) {
		StringBuilder sb = new StringBuilder();
		sb.append("select a, current_timestamp() from ");
		sb.append(BatchInst.class.getName());
		sb.append(" a where catalog='").append(catalog);
		sb.append("' order by addr asc");
		List<Object[]> list = dao.createQuery(sb.toString()).list();
		return list;
	}

	@SuppressWarnings("unchecked")
	@Override
	@Transactional(readOnly = true)
	public BatchInst getActiveLock(int timeout) {
		StringBuilder sb = new StringBuilder();
		sb.append("select a, current_timestamp() from ");
		sb.append(BatchInst.class.getName());
		sb.append(" a order by lockTime desc, addr asc");
		List<Object[]> list = dao.createQuery(sb.toString()).list();
		if (list.isEmpty()) {
			log.warn("No record found in batch_inst: {}", sb);
			return null;
		}
		Object[] oa = list.get(0);
		BatchInst lastLock = (BatchInst) oa[0];
		Timestamp currTimestamp = (Timestamp) oa[1];
		if (currTimestamp.getTime() - lastLock.getLockTime().getTime() <= timeout * 1000l) {
			return lastLock;
		}
		return null;
	}

	@Override
	@Transactional
	public void updateLastTime(String addr) {
		StringBuilder sb = new StringBuilder();
		sb.append("update ").append(BatchInst.class.getName());
		sb.append(" set lastTime = current_timestamp() where addr = '");
		sb.append(addr).append("'");
		dao.executeHql(sb.toString());
	}

	@Override
	@Transactional
	public boolean acquireLock(String addr, int timeout, BatchLock optLock) {
		// 锁定BatchLock：若失败，说明已有其他批量实例更新了BatchInst.lockTime，结束
		if (!batchLockBiz.optimisticLock(optLock, addr)) {
			return false;
		}
		/*
		 * 通过BatchLock.version判断BatchInst.lockTime是否为最新 // 查询当前有效的锁（未超时的锁）
		 * BatchInst activeLock = getActiveLock(timeout); // 检查：是否存在有效的锁 且
		 * 锁不属于自己 if (activeLock != null && !addr.equals(addr)) return false;
		 */
		// 更新LockTime
		StringBuilder sb = new StringBuilder();
		sb.append("update ").append(BatchInst.class.getName());
		sb.append(" set lockTime = current_timestamp(), lastTime = current_timestamp() where addr = '");
		sb.append(addr).append("'");
		dao.executeHql(sb.toString());

		return true;
	}

	@Override
	@Transactional
	public void lock(BatchInst batchInst) {
		// 锁定BatchLock，排除其他调度更新BatchInst.lockTime
		batchLockBiz.pessimisticLock(batchInst.getCatalog(), batchInst.getAddr());

		// sleep 1s
		// 因为可能出现：
		// 原主调度的lastLockTime与本次更新的新主调度的lastLockTime相同，造成，原主调度再次checkLock时锁依然属于自己并再次启动
		// 进程休眠1秒，用来保证新主调度的lastLockTime必定为最大
		try {
			Thread.sleep(1000L);
		} catch (InterruptedException e) {
			throw new RuntimeException("Current thread is interrupted", e);
		}

		// 更新LockTime
		StringBuilder sb = new StringBuilder();
		sb.append("update ").append(BatchInst.class.getName());
		sb.append(" set lockTime = current_timestamp(), lastTime = current_timestamp() where addr = '");
		sb.append(batchInst.getAddr()).append("'");
		dao.executeHql(sb.toString());
	}

	@Override
	@Transactional(readOnly = true)
	public BatchInst get(String addr) {
		return dao.get(BatchInst.class, addr);
	}

	@Override
	@Transactional
	public Object save(BatchInst o) {
		if (Util.strIsEmpty(o.getAddr())) {
			throw new RuntimeException("addr cannot be null");
		} else {
			return dao.merge(o);
		}
	}

	@Override
	@Transactional
	public void delete(String id) {
		dao.delete(BatchInst.class, id);
	}

	@Override
	public List<BatchInst> list() {
		return dao.list(BatchInst.class, " order by lockTime desc, addr asc");
	}
}
